package com.chb.catRecommend;

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 求取同现矩阵
i100:i100	3
i100:i105	1
i100:i106	1
i100:i109	1
i100:i114	1
i100:i124	1
 */
public class Step3 {
	public static boolean run(Configuration conf, Map<String, String> paths) throws Exception {
		FileSystem fs = FileSystem.get(conf);
		Job job = Job.getInstance();
		job.setJar("C:\\Users\\12285\\Desktop\\cr.jar");
		job.setJarByClass(Step3.class);
		job.setJobName("Step3");

		job.setMapperClass(Step3Mapper.class);
		job.setReducerClass(Step3Reducer.class);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);

		Path in = new Path(paths.get("step3Input"));
		FileInputFormat.addInputPath(job, in);
		Path out = new Path(paths.get("step3Output"));
		if (fs.exists(out)) {
			fs.delete(out, true);
		}
		FileOutputFormat.setOutputPath(job, out);
		boolean f = job.waitForCompletion(true);
		return f;
	}
	static class Step3Mapper extends Mapper<LongWritable, Text, Text, IntWritable> {
		//u26	i276:1,i201:1,i348:1,i321:1,i136:1,
		//只需要统计物品两两同现的次数
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String[] vs = value.toString().split("\t")[1].split(",");
			for (int i = 1; i < vs.length; i++) {
				for (int j = i+1; j < vs.length; j++) {
					String itemI = vs[i].split(":")[0];
					String itemJ = vs[j].split(":")[0];
					String K = itemI+":"+itemJ;
					context.write(new Text(K), new IntWritable(1));
				}
			}
		}
	}
	static class Step3Reducer extends Reducer<Text, IntWritable, Text, IntWritable> {
		@Override
		protected void reduce(Text key, Iterable<IntWritable> values, Context context)
				throws IOException, InterruptedException {
			int sum = 0;
			for (IntWritable iw : values) {
				sum += iw.get();
			}
			context.write(key, new IntWritable(sum));
		}
	}
}
